package com.lagou.client;

import com.alibaba.fastjson.JSON;
import com.lagou.codec.JSONSerializer;
import com.lagou.codec.RpcDecoder;
import com.lagou.codec.RpcEncoder;
import com.lagou.protocol.NodeInfo;
import com.lagou.protocol.PathScore;
import com.lagou.protocol.RpcRequest;
import com.lagou.protocol.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class RpcConsumer {

    // 创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    // 创建结果池
    ConcurrentHashMap<String, RpcResponse> resultMap = new ConcurrentHashMap<>();
    // 客户端对象
    Bootstrap bootstrap;
    // 创建channel池
    ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();
    // zk客户端
    CuratorFramework curatorClient;

    public RpcConsumer() throws Exception {
        initClient();
        connectZk();
    }

    //1.创建一个代理对象 rpcRequest
    public Object createProxy(final Class<?> serviceClass){
        //借助JDK动态代理生成代理对象
        return  Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 获取channel -- 主要逻辑
                List<PathScore> collect = channelMap.keySet().stream().map(this::setPathScore).
                        sorted(Comparator.comparingInt(PathScore::getScore)).collect(Collectors.toList());
                System.out.println(collect);

                // 请求数据, 如果没有请求到就返回
                for (PathScore p: collect) {
                    try {
                        String path = p.getPath();
                        Channel channel = channelMap.get(path);

                        // 构造请求体
                        ClientRequestHandler clientRequestHandler = new ClientRequestHandler(resultMap, channel);
                        clientRequestHandler.setRequest(createRpcRequest(serviceClass, method, args));

                        // 去服务端请求数据
                        System.out.println("path: " + path + " process this request");
                        String s = executor.submit(clientRequestHandler).get();

                        // 不为null才返回，不然去下一个请求，全空由业务决定处理流程
                        if (s != null) {
                            return s;
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return null;
            }

            private PathScore setPathScore(String s) {
                PathScore pathScore = new PathScore();
                pathScore.setPath(s);

                try {
                    byte[] bytes = curatorClient.getData().forPath(s);
                    if (bytes != null && bytes.length > 0) {
                        NodeInfo nodeInfo = JSON.parseObject(bytes, NodeInfo.class);
                        int time = nodeInfo.getResponseTime();
                        System.out.println("set " + s + " score: " + time);
                        long lastModified = nodeInfo.getLastModified();
                        long currentTime = (new Date()).getTime();
                        if (currentTime - lastModified < 5000) {
                            pathScore.setScore(time);
                        } else {
                            System.out.println("path: " + s + " expire, set to 0");
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return pathScore;
            }
        });
    }

    private RpcRequest createRpcRequest(Class<?> aClass, Method method, Object[] args) {
        RpcRequest rpcRequest = new RpcRequest();
        rpcRequest.setClassName(aClass.getName());
        rpcRequest.setMethodName(method.getName());
        rpcRequest.setParameters(args);
        rpcRequest.setParameterTypes(method.getParameterTypes());
        return rpcRequest;
    }

    // 初始化netty客户端
    public void initClient() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        final JSONSerializer jsonSerializer = new JSONSerializer();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, jsonSerializer)); // encoder设置为RpcRequest+json
                        pipeline.addLast(new RpcDecoder<>(RpcResponse.class, jsonSerializer));
                        pipeline.addLast(new ClientResponseHandler(resultMap));
                    }
                });
    }

    // 初始化和zk的连接
    public void connectZk() throws Exception {
        String zookeeperAddress = "192.168.200.200:2181";

        // 启动zookeeper
        curatorClient =
                CuratorFrameworkFactory.builder().
                        connectString(zookeeperAddress) //连接地址
                        .connectionTimeoutMs(2000) //连接超时时间
                        .sessionTimeoutMs(10000) //会话超时时间
                        .retryPolicy(new ExponentialBackoffRetry(1000, 3)) //重试策略
                        .namespace("rpc_manage") //命名空间,默认节点
                        .build();
        curatorClient.start();
        // 注册监听
        addChildrenListener(curatorClient);

        // 初始化channel池
        List<String> childrenList = curatorClient.getChildren().forPath("/");
        for(String node: childrenList) {
            String path = "/" + node;
            byte[] bytes = curatorClient.getData().forPath(path);
            addChannel(bytes, path);
        }
    }

    public void addChildrenListener(CuratorFramework curatorClient) throws Exception {
        // 注册监听
        PathChildrenCache childrenCache = new PathChildrenCache(curatorClient, "/", true);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                ChildData data = pathChildrenCacheEvent.getData();
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                System.out.println("event type: " + type);
                if (type == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                    addChannel(data.getData(), data.getPath());
                } else if (type == PathChildrenCacheEvent.Type.CHILD_REMOVED){
                    removeChannel(data.getPath());
                } else {
                    System.out.println("unknown type: " + type);
                }
            }
        };
        childrenCache.getListenable().addListener(pathChildrenCacheListener);
        childrenCache.start();
    }

    private synchronized void removeChannel(String path) {
        if (channelMap.containsKey(path)) {
            Channel channel = channelMap.get(path);
            channel.close();
            channelMap.remove(path);
            System.out.println(channelMap);
        } else {
            System.out.println("channel for path: " + path + " doesn't exist");
        }
    }

    private synchronized void addChannel(byte[] bytesData, String path) throws InterruptedException {
        if (channelMap.containsKey(path)) {
            System.out.println("channel for path: " + path + " already exist");
        } else {
            NodeInfo nodeInfo = JSON.parseObject(bytesData, NodeInfo.class);
            String ip = nodeInfo.getIp();
            int port = nodeInfo.getPort();
            Channel channel = bootstrap.connect(ip, port).sync().channel();
            channelMap.put(path, channel);
            System.out.println(channelMap);
        }
    }
}
